package com.baishancloud.log.live.spider

import cn.hutool.json.JSONUtil
import com.baishancloud.log.common.env.StreamEnv
import com.baishancloud.log.common.sink.SinkUtil
import com.baishancloud.log.common.source.SourceUtil
import com.baishancloud.log.format.scala.SpiderOrigin
import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.slf4j.{Logger, LoggerFactory}

import java.io.Serializable
import java.util.Objects
import scala.language.postfixOps

class StreamingSpiderLive extends Serializable

object StreamingSpiderLive extends Serializable {

  val LOGGER: Logger = LoggerFactory.getLogger(classOf[StreamingSpiderLive])


  def main(args: Array[String]): Unit = {
    val parameterTool = ParameterTool.fromArgs(args)
    println("输入参数为：" + parameterTool.toMap.toString)
    StreamEnv.builder(parameterTool).build()
    val source1: DataStream[String] = SourceUtil.kafkaOnlyValue(parameterTool, "1")
    val source2: DataStream[String] = SourceUtil.kafkaOnlyValue(parameterTool, "2")
    val source: DataStream[String] = source1.union(source2)
    source
      .map(str => {
        try {
          SpiderOrigin.parse(str).orNull
        } catch {
          case _: Exception => {
            null
          }
        }
      }).name("SpiderOrigin").uid("6afef8b-6212-441d-b98a-8f4488a2a5e1")
      .filter(_ != null).name("!null").uid("e5d8d26-81af-4551-ba0e-b103afa62781")
      .filter(x => {
        val current: Long = System.currentTimeMillis() / 1000
        val allowOldest: Long = current - 3600 * parameterTool.getLong(allowOldestHour, 6)
        val logTime: Long = x.datetimeOfUTC().toEpochSecond
        //日志时间在6小时以内
        val b1: Boolean = logTime > allowOldest
        //${connect-status}值为1（即正在拉流）
        val b2: Boolean = Objects.equals(x.connectStatus(), "1")
        //${x-comment}值为-，取边缘
        val b3 = Objects.equals(x.xComment(), "-")
        b1 && b2 && b3
      }).name("someFilter").uid("4e422ca-80e5-4cbf-8e04-f5409804486b")
      .process(new SpiderLogRecordParse(parameterTool)).name("LogRecordParse").uid("d363d75-afe2-4401-bb0f-03205e019601")
      .keyBy(_.getKey)
      .window(TumblingProcessingTimeWindows.of(Time.minutes(parameterTool.getLong(windowSize, 1))))
      .process(new SpiderWindowAgg(parameterTool)).name("uvAgg").uid("c158e29-7fea-421e-acd8-f00e5d727375")
      .map(JSONUtil.toJsonStr(_)).name("toJsonStr").uid("d58a9b82-024b-487-80ab-81004be9974d")
      .addSink(SinkUtil.starRocksJsonString(parameterTool)).setParallelism(parameterTool.getInt(sinkParallel, 1)).name("starRocks").uid("465e98a-f3b0-4ad9-b6e6-a742faa4c393")
    StreamEnv.execute(parameterTool)
  }

}